package com.ding.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class AmqpConfig {

	public AmqpConfig(RabbitTemplate rabbitTemplate) {

		//当配置confirms为true时该接口无论是否将消息发送到交换机上都会被调用.而如果为false则只会在发送失败时调用
        rabbitTemplate.setConfirmCallback(

                (correlationData, ack, cause) -> {
                    System.out.println("发送确认.只确认有没有路由到交换机:"+ack);
                }

        );

		

		//当配置returns为true时,该接口会在由交换机路由至队列失败时调用
        rabbitTemplate.setReturnCallback(
                (message, replyCode, replyText, exchange, routingKey) -> {
                    System.out.println("消息被退回:"+exchange);
                }
        );

    }
	//创建消息队列
	@Bean
	public Queue queue_a() {
		return new Queue("a");
	}
	@Bean
	public Queue queue_b() {
		return new Queue("b");
	}

	//___________交换机__________
	
	/*
	 * 交换机或称路由器,是用来将消息按照不同的策略向目的地路由的中转站
	 * 路由策略:
	 * 		1.FanoutExchange:广播交换机,无所谓目的地,所有绑定到该交换机的队列都会收到消息
	 * 		2.DirectExchange:直接交换机(完全匹配交换机),不支持模糊匹配,消息目的地必须和绑定的routingKey完全一致
	 * 		3.TopicExchange	:主题交换机,支持模糊匹配,"*"模糊匹配一个单词,"#"模糊匹配任意个单词
	 */
	@Bean
	public FanoutExchange fanoutExchange() {
		return new FanoutExchange("fanout");
	}
	
	@Bean
	public DirectExchange directExchange() {
		return new DirectExchange("direct");
	}
	@Bean
	public TopicExchange topicExchange() {
		return new TopicExchange("topic");
	}
	
	//_________绑定器_______________
	
	/*
	 * 使用routingKey将队列绑定到交换机
	 */
	
	@Bean
	public Binding a_fanoutExchangeBinding() {
		return BindingBuilder.bind(queue_a()).to(fanoutExchange());
	}
	@Bean
	public Binding b_fanoutExchangeBinding() {
		return BindingBuilder.bind(queue_b()).to(fanoutExchange());
	}
	@Bean
	public Binding b_directExchangeBinding() {
		return BindingBuilder.bind(queue_b()).to(directExchange()).with("direct.b");
	}
	@Bean
	public Binding a_topicExchangeBinding() {
		return BindingBuilder.bind(queue_a()).to(topicExchange()).with("topic.*");

	}
	@Bean
	public Binding b_topicExchangeBinding() {
		return BindingBuilder.bind(queue_b()).to(topicExchange()).with("topic.#");
	}










		//_______延时消息________
	
	

    //缓冲队列
    @Bean
    public Queue cq() {

        return QueueBuilder.durable("fq")	
                .withArgument("x-message-ttl",20000L)	//队列中消息的有效时长
                .withArgument("x-dead-letter-exchange", "e")	//死信转发到的交换机
                .withArgument("x-dead-letter-routing-key", "a")	//死信转发时使用的路由键
                .build();
    }

	//死信队列
	@Bean
    public Queue aq() {
        return new Queue("aq", true);
    }

	//死信交换机
    @Bean
    public DirectExchange e(){
        return new DirectExchange("e");
    }

    @Bean
    public Binding b() {
        return BindingBuilder.bind(aq())
                .to(e())
                .with("a");
    }

    @RabbitListener(queues = "aq")
    public void test1 ()
    {
        System.out.println(System.currentTimeMillis());
    }


	@Bean //默认的消息转换器只能发送String,Jackson2JsonMessageConverter则可以发送一般的对象
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }

    @RabbitListener(queues = "test")
    public void hand(Message message,@Header(AmqpHeaders.DELIVERY_TAG) long tag, Channel channel) throws IOException{

		//转换消息
        Object o = messageConverter().fromMessage(message);

        System.out.println(o.getClass());

        channel.basicAck(tag, false);
    }


}
